//
// Created by guo on 24-8-30.
//

#include "BBClient.h"
#include "protocol.h"

bool BBClient::connect(const std::string &ip, uint16_t port) {
    _exit.store(false);
    if(!_client){
        _client = new TCPClient;
        if(!_client->connect(ip,port)){
            return false;
        }
        _t = std::thread([this](){
            run();
        });
        return true;
    }else{
        _exit.store(true);
        return false;
    }
}

void BBClient::run() {
    std::thread handle_thread(&BBClient::handle_data,this);
    std::thread poll_thread(&BBClient::poll_one,this);

    poll_thread.join();
    handle_thread.join();
}

void BBClient::set_blackboard(const std::string &key, const std::string &value) {
    std::string raw_data;
    BBMsg::gen_raw_data(BBMsgType::SET,BBMsgError::NO_ERROR,key,raw_data,value);
    _client->send(raw_data);
}

bool BBClient::get_blackboard(const std::string &key, std::string &value) {
    std::string raw_data;
    BBMsg::gen_raw_data(BBMsgType::GET,BBMsgError::NO_ERROR,key,raw_data);
    _client->send(raw_data);
    bool ret = Blackboard::wait_memory_bb(key,value);
    return (ret && (!value.empty()));
}

void BBClient::handle_data() {
    std::vector<std::pair<BBMsg,std::string>> msgs;
    while(!_exit.load()){
        QueueDataPtr queue_data;
        bool ret = _client->get_data(queue_data);
        if(ret){
            parse_data(queue_data->data,msgs);
            handle_msg(msgs);
        }else{
            std::this_thread::sleep_for(std::chrono::milliseconds(80));
        }
    }
}

void BBClient::poll_one(){
    while(!_exit.load()){
        _client->poll_one();
    }
}

void BBClient::parse_data(const std::string &data, std::vector<std::pair<BBMsg,std::string> > &vec_msg) {
    vec_msg.clear();
    if(data.size() < sizeof(BBMsg)){
        return;
    }

    std::string header;
    header.resize(2);
    header[0] = 0xEF;
    header[1] = 0x06;

    size_t pos = 0;
    size_t header_pos = 0;
    header_pos = data.find(header,pos);
    if(header_pos == std::string::npos){
        return;
    }
    std::string msg_data = data.substr(header_pos,sizeof(BBMsg));
    BBMsg* msg = (BBMsg*)msg_data.data();
    auto pair = std::make_pair(*msg,std::string(data.data()+header_pos+sizeof(BBMsg),msg->len));
    vec_msg.push_back(pair);
}

void BBClient::handle_msg(std::vector<std::pair<BBMsg, std::string>> &vec_msg) {
    for(const auto& bbmsg_pair : vec_msg){
        switch (static_cast<BBMsgType>(bbmsg_pair.first.type)) {
            case BBMsgType::UNKNOWN:
                break;
            case BBMsgType::SET:
                break;
            case BBMsgType::GET:{
                std::string key_str(reinterpret_cast<const char *>(bbmsg_pair.first.key),
                                    strlen(reinterpret_cast<const char *>(bbmsg_pair.first.key)));
                if(static_cast<BBMsgError>(bbmsg_pair.first.error_code) == BBMsgError::NO_ERROR){
                    Blackboard::notify_memory_bb(key_str,bbmsg_pair.second);
                }else{
                    Blackboard::notify_memory_bb(key_str,"");
                }
                break;
            }
            case BBMsgType::ERASE:
                break;
        }
    }
}

void BBClient::erase_blackboard(const std::string &key) {
    std::string raw_data;
    BBMsg::gen_raw_data(BBMsgType::ERASE,BBMsgError::NO_ERROR,key,raw_data);
    _client->send(raw_data);
}

BBClient::~BBClient() {
    _t.join();
}
